package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App10_DwdMhsCarePlan extends BaseSqlApp {
    public static void main(String[] args) {
        new App10_DwdMhsCarePlan().init(
                5010,
                "App10_DwdMhsCarePlan",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "RECORD_ID                    String,"+
                "HOSPITAL_NAME                String,"+
                "HOSPITAL_CODE                String,"+
                "AGE_UNIT                     String,"+
                "AGE                          String,"+
                "SEX_NAME                     String,"+
                "SEX_CODE                     String,"+
                "ID_CARD_NUMBER               String,"+
                "PATIENT_VISIT_CATEGORY_CODE  String,"+
                "PATIENT_VISIT_CATEGORY       String,"+
                "INPATIENT_VISIT_FLOW_ID      String,"+
                "INPATIENT_NO                 String,"+
                "PATIENT_ID                   String,"+
                "PATIENT_NAME                 String,"+
                "DOMAIN_ID                    String,"+
                "VISIT_DATE                   String,"+
                "BED_NAME                     String,"+
                "ROOM_NAME                    String,"+
                "WARD_ID                      String,"+
                "VISIT_TIMES                  String,"+
                "HEALTH_CARD                  String,"+
                "BED_NO                       String,"+
                "ROOM_NO                      String,"+
                "DEPT_ID                      String,"+
                "DEPT_NAME                    String,"+
                "WARD_NAME                    String,"+
                "AUTHOR_CODE                  String,"+
                "AUTHOR_NAME                  String,"+
                "CUSTODIAN_CODE               String,"+
                "CUSTODIAN_NAME               String,"+
                "DOC_AUDIT_DATE               String,"+
                "DOC_AUDITORCODE              String,"+
                "DOC_AUDITOR                  String,"+
                "DIAG_CODE                    String,"+
                "DIAG_NAME                    String,"+
                "NURSING_GRADECODE            String,"+
                "NURSE_GRADE_NAME             String,"+
                "NURSING_TYPECODE             String,"+
                "NURSE_TYPE_NAME              String,"+
                "NURSING_PROBLEM              String,"+
                "NURSE_NAME                   String,"+
                "NURSE_CATEGORY_ITEM          String,"+
                "NURSE_RESULT                 String,"+
                "CATHETER_DESCRIPTION         String,"+
                "POSTURE_NURSING              String,"+
                "SKIN_CARE                    String,"+
                "TRACHEA_CODE                 String,"+
                "TRACHEA_NAME                 String,"+
                "SAFETY_CODE                  String,"+
                "SAFETY_NAME                  String,"+
                "GUIDE_CODE                   String,"+
                "GUIDE_NAME                   String," +
                "VISIT_ID                     String,"+
                "primary key (RECORD_ID) not enforced" +
                ")"+
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_CARE_PLAN));
        tableEnv.executeSql("insert into target select * from join_result");
    }


    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id             AS RECORD_ID," +
                "hospital_name         AS HOSPITAL_NAME," +
                "hospital_code         AS HOSPITAL_CODE," +
                "age_unit              AS AGE_UNIT," +
                "age                   AS AGE," +
                "sex                   AS SEX_NAME," +
                "sex_code              AS SEX_CODE," +
                "identify_no           AS ID_CARD_NUMBER," +
                "patient_typecode      AS PATIENT_VISIT_CATEGORY_CODE," +
                "patient_type          AS PATIENT_VISIT_CATEGORY," +
                "visit_id              AS INPATIENT_VISIT_FLOW_ID," +
                "inpatient             AS INPATIENT_NO," +
                "patient_id            AS PATIENT_ID," +
                "patient_name          AS PATIENT_NAME," +
                "s.domain_id             AS DOMAIN_ID," +
                "visit_date            AS VISIT_DATE," +
                "bed_name              AS BED_NAME," +
                "ward_name             AS ROOM_NAME," +
                "wards_id              AS WARD_ID," +
                "visit_times           AS VISIT_TIMES," +
                "health_card           AS HEALTH_CARD," +
                "bed_no                AS BED_NO," +
                "ward_id               AS ROOM_NO," +
                "dept_code             AS DEPT_ID," +
                "dept_name             AS DEPT_NAME," +
                "wards_name            AS WARD_NAME," +
                "author_code           AS AUTHOR_CODE," +
                "author_name           AS AUTHOR_NAME," +
                "custodian_code        AS CUSTODIAN_CODE," +
                "custodian_name        AS CUSTODIAN_NAME," +
                "doc_audit_date        AS DOC_AUDIT_DATE," +
                "doc_auditorcode       AS DOC_AUDITORCODE," +
                "doc_auditor           AS DOC_AUDITOR," +
                "diag_code             AS DIAG_CODE," +
                "diag_name             AS DIAG_NAME," +
                "nursing_gradecode     AS NURSING_GRADECODE," +
                "nursing_grade         AS NURSE_GRADE_NAME," +
                "nursing_typecode      AS NURSING_TYPECODE," +
                "nursing_type          AS NURSE_TYPE_NAME," +
                "nursing_problem       AS NURSING_PROBLEM," +
                "nurse_name            AS NURSE_NAME," +
                "nurse_category_item   AS NURSE_CATEGORY_ITEM," +
                "nurse_result          AS NURSE_RESULT," +
                "catheter_description  AS CATHETER_DESCRIPTION," +
                "posture_nursing       AS POSTURE_NURSING," +
                "skin_care             AS SKIN_CARE," +
                "c.code                AS TRACHEA_CODE," +
                "c.name                AS TRACHEA_NAME," +
                "b.code                AS SAFETY_CODE," +
                "b.name                AS SAFETY_NAME," +
                "a.code                AS GUIDE_CODE," +
                "a.name                AS GUIDE_NAME," +
                "visit_id              AS VISIT_ID " +
                "from hdsd00_09_03 s " +
                "left join tdy_list_diet_guide a  on a.unique_id=s.unique_id and a.xds_type=s.xds_type  " +
                "left join tdy_list_nurse_safety b  on b.unique_id=s.unique_id and b.xds_type=s.xds_type  " +
                "left join tdy_list_nurse_trachea c on c.unique_id=s.unique_id and c.xds_type=s.xds_type  " +
                "");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table hdsd00_09_03 (" +
                "pk                    String,"+
                "upload_time           String,"+
                "status                String,"+
                "empi                  String,"+
                "encounter_id          String,"+
                "visit_date            String,"+
                "bed_name              String,"+
                "ward_name             String,"+
                "wards_id              String,"+
                "visit_domain          String,"+
                "visit_id              String,"+
                "visit_times           String,"+
                "patient_id            String,"+
                "patient_domain        String,"+
                "unique_id             String,"+
                "xds_type              String,"+
                "xds_name              String,"+
                "effective_time        String,"+
                "xds_version           String,"+
                "domain_id             String,"+
                "patient_type          String,"+
                "patient_typecode      String,"+
                "health_card           String,"+
                "inpatient             String,"+
                "identify_no           String,"+
                "patient_name          String,"+
                "sex_code              String,"+
                "sex                   String,"+
                "age_unit              String,"+
                "age                   String,"+
                "bed_no                String,"+
                "ward_id               String,"+
                "dept_code             String,"+
                "dept_name             String,"+
                "wards_name            String,"+
                "hospital_code         String,"+
                "hospital_name         String,"+
                "write_time            String,"+
                "author_code           String,"+
                "author_name           String,"+
                "custodian_code        String,"+
                "custodian_name        String,"+
                "doc_audit_date        String,"+
                "doc_auditorcode       String,"+
                "doc_auditor           String,"+
                "diag_code             String,"+
                "diag_name             String,"+
                "nursing_gradecode     String,"+
                "nursing_grade         String,"+
                "nursing_typecode      String,"+
                "nursing_type          String,"+
                "nursing_problem       String,"+
                "nurse_name            String,"+
                "nurse_category_item   String,"+
                "nurse_result          String,"+
                "catheter_description  String,"+
                "posture_nursing       String,"+
                "skin_care             String "+
                ")"+
                FlinkSourceUtil.getKafkaWith("hdsd00_09_03","App10_DwdMhsCarePlan"));

        super.readTdyListNurseTrachea(tableEnv,"App10_DwdMhsCarePlan");
        super.readTdyListNurseSafety(tableEnv,"App10_DwdMhsCarePlan");
        super.readTdyListDietGuide(tableEnv,"App10_DwdMhsCarePlan");
    }
}
